package com.gdlt.mq2db;

import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.hadoop.conf.Configuration;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class GraphWriter extends Thread implements Writable {
	
	Configuration conf = Kafka2pgConfig.getConf();
	String graphHost = conf.get(Kafka2pgConfig.GRAPHITE_HOST_CONF_NAME);
	int graphitePort = conf.getInt(Kafka2pgConfig.GRAPHITE_PORT_CONF_NAME, Kafka2pgConfig.GRAPHITE_PORT_DEFAULT_VALUE);
	Socket socket = null;
	OutputStream stream = null;
	PrintWriter out = null;
	long offset;

	BlockingQueue<ConsumerRecord<String, String>> queue;
	private static AtomicBoolean flag = new AtomicBoolean(true);
	 private final static Logger LOG = LoggerFactory.getLogger(GraphWriter.class);
	
	public GraphWriter(BlockingQueue<ConsumerRecord<String, String>> queue) {
		this.queue = queue;
		try {
			socket = createSocket();
			stream = socket.getOutputStream();
			//LOG.info("连接到主机： "+graphHost);
		} catch (UnknownHostException e) {
			// TODO Auto-generated catch block
			LOG.error("主机名出错： "+e);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} 
	}

	public void write(List<ConsumerRecord<String, String>> data,long offset) {
		// TODO Auto-generated method stub

		List<String> list = new ArrayList<String>();
		for (int i = 0; i < data.size(); i++) {
			String recValue = data.get(i).value(); // 值里面就包括时间戳
			list.add(recValue);

		}
		if (socket.isClosed() == false && socket.isConnected() == true) {
			out = new PrintWriter(stream, true);
			out.println(list);//PrintWriter的println函数
		} else {
			LOG.error("连接断开,尝试重连......");
			try {
				socket = createSocket();
				stream = socket.getOutputStream();
				LOG.info("重连成功");
			} catch (UnknownHostException e) {
				// TODO Auto-generated catch block
				LOG.error("主机名出错："+e);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				LOG.error("写数据 到Graphite时出错: " +e);
			}
		}
	}
	


	// 创建连接
	protected Socket createSocket() throws UnknownHostException, IOException {
		return new Socket(graphHost, graphitePort);

	}

	public boolean isClosed() {
		// TODO Auto-generated method stub
		return flag.get();
	}

	public void close() {
		// TODO Auto-generated method stub
		try {
			if (flag.compareAndSet(true, false)) {
			out.close();
			socket.close();
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public void run() {
		// TODO Auto-generated method stub
		System.out.println( " GraphWriter id is " + Thread.currentThread().getId());
		List<ConsumerRecord<String, String>> data = new ArrayList<ConsumerRecord<String, String>>();
		try {
			while (true) {
				data.add(queue.take());
				if (data.size() >= 10000) {
					write(data,offset);
					data.clear();
				}
			}

		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}
